package server

import (
	"context"
	"github.com/go-kratos/kratos/v2/log"
	"github.com/tx7do/kratos-transport/transport/kafka"
	"kratos_kafka_new/internal/conf"
	"kratos_kafka_new/internal/service"
)

func NewKafkaConsumerServer(msgConf *conf.MessageQueue, greeterService *service.GreeterService) (*kafka.Server, func()) {
	// Notice TLS～～～
	var err error
	//certBytes, err := ioutil.ReadFile(msgConf.Kafka.Pem)
	//if err != nil {
	//	panic(fmt.Sprintf("kafka client read cert file failed %v", err))
	//}
	//clientCertPool := x509.NewCertPool()
	//ok := clientCertPool.AppendCertsFromPEM(certBytes)
	//if !ok {
	//	panic("kafka client failed to parse root certificate")
	//}

	ctx := context.Background()
	kafkaSrv := kafka.NewServer(
		kafka.WithAddress(msgConf.Kafka.Server.Addr), // addr是个列表
		// Notice kratos-transport/transport 1.0.5版本的方法
		kafka.WithCodec("json"),

		// TLS～～～
		//kafka.WithTLSConfig(&tls.Config{
		//	RootCAs:            clientCertPool,
		//	InsecureSkipVerify: true,
		//}),
		//kafka.WithPlainMechanism(c.Kafka.Username, c.Kafka.Password),
		////kafka.WithBrokerOptions(broker.OptionContextWithValue("max.poll.records", 5000)),
	)

	// 注册一个订阅者，从kafka中消费数据
	//err = kafkaSrv.RegisterSubscriber(
	//	ctx,
	//	msgConf.Kafka.ArticleTopic, // topic
	//	msgConf.Kafka.ArticleGroup, // group
	//	true,
	//	mq_kafka.RegisterArticleHandler2(greeterService.HandleReceiveArticleData), // article_group1
	//	mq_kafka.DataCreator,
	//)
	if err != nil {
		panic(err)
	}

	// Notice 注册另外一个订阅者，topic跟上面的一样，用另外一个group~
	//err = kafkaSrv.RegisterSubscriber(
	//	ctx,
	//	msgConf.Kafka.ArticleTopic,
	//	msgConf.Kafka.ArticleGroup2,
	//	false,
	//	mq_kafka.RegisterArticleHandler(greeterService.HandleReceiveArticleData2), // article_group2
	//	mq_kafka.DataCreator,
	//)
	//if err != nil {
	//	panic(err)
	//}

	// start
	if err = kafkaSrv.Start(ctx); err != nil {
		panic(err)
	}

	return kafkaSrv, func() {
		if err = kafkaSrv.Stop(ctx); err != nil {
			log.Errorf("kafka expected nil got %v", err)
		}
	}
}
